package com.example.bootconsumer.listeners;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

@Component
public class BootRabbitMQListener {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @RabbitListener(queues = "boot.topicQue1")
    public void topicQue1Consumer(Message msg){
        System.out.println("que1消费者");

        byte[] body = msg.getBody();
        String b = new String(body, StandardCharsets.UTF_8);
        System.out.println("消息： " + b);

        System.out.println("-------------------");
    }

    /**
     * springboot消息消费时，是自动提交的。如果消费过程中不报异常，就认为消息消费成功，提交-消息就会从服务器删除
     *                                  如果消息消费过程中出现了异常，springboot会进行重试
     *
     *              问题：
     *                  ① 无限重试 -- 可以通过redis控制重试次数
     *                  ② 实际的业务处理的结果是否成功，并不一定会通过异常来标识
     *                      比如有一个业务：返回1表示成功，返回2表示失败
     *
     *             解决方案：
     *                  改成手动提交
     *                      ① 在yml中进行全局配置
     *                      ② 在某个消费者上单独配置（这个优先级比全局优先级高）
     * @param msg
     */
    @RabbitListener(queues = "boot.topicQue2"/*,ackMode = "MANUAL"*/)
    public void topicQue2Consumer(Message msg, Channel channel) throws IOException {
        System.out.println("que2消费者");

        try {

            byte[] body = msg.getBody();
            String b = new String(body, StandardCharsets.UTF_8);
            System.out.println("消息： " + b);

            System.out.println("-------------------");

            //确认
            channel.basicAck(msg.getMessageProperties().getDeliveryTag(),false);
        }catch (Exception e){
            //重试 -- 控制重试次数
            channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,true);
        }
    }

//    @RabbitListener(queues = "normal.topicQue1")
    public void topicQue3Consumer(Message msg,Channel channel) throws IOException {
        System.out.println("que3消费者");

        byte[] body = msg.getBody();
        String b = new String(body, StandardCharsets.UTF_8);
        System.out.println("消息： " + b);

        System.out.println("-------------------");

        //拒绝不入队
        channel.basicNack(msg.getMessageProperties().getDeliveryTag(),false,false);
    }

    @RabbitListener(queues = "deadletter.topicQue1",ackMode = "AUTO")
    public void topicQue4Consumer(Message msg,Channel channel) throws IOException {
        System.out.println("que4消费者");

        byte[] body = msg.getBody();
        String b = new String(body, StandardCharsets.UTF_8);
        System.out.println("消息： " + b);

        System.out.println("-------------------" + LocalDateTime.now());
    }

    /**
        增：非幂等
            业务流程：先【查询】数据库是否有记录，如果没有，再【新增】，必须让【查询】和【新增】是一个原子性的

            ① 分布式锁，锁键用消息的唯一标识 （setnx/redisson）
            ② 可以使用数据库自身的锁机制 (保障查询和插入操作必须在同一个事务中)
                select * from xxx where 条件=值 for update;
                查不到 insert ....
            ③ 直接利用数据库的唯一约束（要记得捕获唯一约束异常）

        删：幂等

        改：
            设置型语句：幂等
                update 表名 set 列=值,... where 条件；

            计算型语句：非幂等
                update 表名 set 列=列+值,... where 条件；

                ① 分布式锁，锁键用消息的唯一标识 （setnx/redisson）
                ② 利用状态机机制
                    update 表名 set 列=列+值,... status='已修改' where 条件 and status='未修改'；

        查：幂等

     */
    @RabbitListener(queues = "mideng.topicQue1",ackMode = "AUTO")
    public void topicQue5Consumer(Message msg) throws IOException {
        String messageId = msg.getMessageProperties().getMessageId();

        Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(messageId, "", 10, TimeUnit.SECONDS);

        if (ifAbsent) {

            try {
                byte[] body = msg.getBody();
                String b = new String(body, StandardCharsets.UTF_8);
                System.out.println("消费者4消息： " + b + " 消息id为：" + messageId);

                System.out.println("-------------------" + LocalDateTime.now());
            }catch (Exception e){
                //重试 -- 要删除原来那个id对应的值
                redisTemplate.delete(messageId);
            }
        }
    }

}
